""" 
    Author: Siren
    Time: 2023/6/20 17:05 
    File: AgeModel_siren.py 
    IDE: PyCharm 
"""
# !/usr/bin/env python
# -*- coding:utf-8 -*-
# 1-Python Anaconda 虚拟环境创建命令
# 虚拟环境如何查看有哪些？conda env list
# 虚拟环境如何创建？conda create -n pyspark_env python==3.7.10
# 虚拟环境如何启动？conda activate pyspark_env
# 环境准备
from pyspark import SparkContext
from pyspark.sql import DataFrame, SparkSession
from pyspark.sql.functions import regexp_replace

from index_development.BaseModel.BaseModelAbstract import BaseModel


class AgeModel(BaseModel):
    # TODO 1.获取年龄段标签id
    def getTagId(self):
        return 8
    # TODO 2.获得newDF
    def compute(self, esDF: DataFrame, fiveDF: DataFrame, sc: SparkSession, spark: SparkContext):
        print('=================清洗5级标签规则=================')
        """将出生日期的时间段(yyyyMMdd-yyyyMMdd)划分成开始(yyyyMMdd)和结束(yyyyMMdd)两个字段"""
        fiveDF_etl:DataFrame= fiveDF.rdd\
            .map(lambda row: (row['id'], row['rule'].split('-')[0], row['rule'].split('-')[1]))\
            .toDF(['tagsId', 'start', 'end'])
        fiveDF_etl.show()
        print('==============清洗客户生日,去除"-"===============')
        esDF_etl:DataFrame= esDF.select(esDF['user_id'].alias('userId'),
                               regexp_replace(esDF['birthday'][0:10], '-', '').alias('birthday'))
        esDF_etl.show()
        print('========将客户生日与年龄段标签规则匹配,打标签========')
        newDF:DataFrame= esDF_etl.join(fiveDF_etl)\
            .where(esDF_etl['birthday'].between(fiveDF_etl['start'], fiveDF_etl['end']))\
            .select(esDF_etl['userId'].cast('string'),fiveDF_etl['tagsId'].cast('string'))
        newDF.show()
        return newDF

if __name__ == '__main__':
    ageModel=AgeModel()
    ageModel.execute()
        